package com.zhang.gmall.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.zhang.gmall.realtime.common.GmallConfig;
import com.zhang.gmall.realtime.utils.DimUtil;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Collection;
import java.util.Set;

/**
 * @title: 自定义写入Phoenix表sink（可以动态根据表名写入多张表）
 * Flink自带的JDBC SINK 只能往一张表中写，所以需要自定义sink
 * @author: zhang
 * @date: 2022/3/5 22:22
 */
public class DimSinkFunc extends RichSinkFunction<JSONObject> {
    private Connection connection;

    @Override
    public void open(Configuration parameters) throws Exception {
        //注册驱动
        Class.forName(GmallConfig.PHOENIX_DRIVER);
        //获取连接
        connection = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
        //在向Phoenix表中，需要手动提交事务，因为Phoenix自动提交事务为false
        connection.setAutoCommit(true);
    }

    @Override
    public void invoke(JSONObject jsonObj, Context context) {
        PreparedStatement statement = null;

        try {
            //获取维度表
            String table = jsonObj.getString("sinkTable");
            //获取变化维度数据
            JSONObject data = jsonObj.getJSONObject("data");
            //生成upsert语句
            String upsertSQL = genUpsertSql(table, data);
            System.out.println("upsert语句：" + upsertSQL);
            statement = connection.prepareStatement(upsertSQL);
            statement.executeUpdate();
            //注意：再向Phoenix表中，需要手动提交事务,或者开启自动提交
            //事务只针对DML；DQL、DDL是没有事务的
            //connection.commit();
            //如果维度数据发生变化删除redis缓存
            if ("update".equals(jsonObj.getString("type"))) {
                DimUtil.deleteCached(table.toUpperCase(), data.getString("id"));
            }

        } catch (SQLException e) {
            e.printStackTrace();
            throw new RuntimeException("向Phoenix表中插入数据失败！");
        } finally {
            if (statement != null) {
                try {
                    statement.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }

    }

    //upsert into student(id,name,sex) values('1001','zhangsan','beijing');
    private String genUpsertSql(String table, JSONObject data) {
        Set<String> keySet = data.keySet();
        Collection<Object> values = data.values();
        StringBuilder upsertSQL = new StringBuilder("upsert into ");
        upsertSQL.append(GmallConfig.HBASE_SCHEMA)
                .append(".")
                .append(table)
                .append("( ")
                .append(StringUtils.join(keySet, ","))
                .append(" ) ")
                .append("values('")
                .append(StringUtils.join(values, "','"))
                .append("')");
        return upsertSQL.toString();
    }

    @Override
    public void close() throws Exception {
        connection.close();
    }
}
